package com.raft.core.log.statemachine;

import java.io.IOException;
import java.io.InputStream;
import java.util.LinkedList;
import java.util.List;
import java.util.SortedMap;
import java.util.TreeMap;

import javax.annotation.Nonnull;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.raft.core.log.snapshot.Snapshot;
import com.raft.core.support.SingleThreadTaskExecutor;
import com.raft.core.support.TaskExecutor;

public abstract class AbstractSingleThreadStateMachine implements StateMachine {

    private static final Logger logger = LoggerFactory.getLogger(AbstractSingleThreadStateMachine.class);
    private volatile int lastApplied = 0;
    private final TaskExecutor taskExecutor;
    /**等待用的数据结构
     */
    private final SortedMap<Integer, List<String>> readIndexMap = new TreeMap<>();

    public AbstractSingleThreadStateMachine() {
        taskExecutor = new SingleThreadTaskExecutor("state-machine");
    }

    @Override
    public int getLastApplied() {
        return lastApplied;
    }

    @Override
    public void applyLog(StateMachineContext context, int index, @Nonnull byte[] commandBytes, int firstLogIndex) {
        taskExecutor.submit(() -> doApplyLog(context, index, commandBytes, firstLogIndex));
    }

    /**应用日志(taskExecutor的线程中)
     * @param context
     * @param index
     * @param commandBytes
     * @param firstLogIndex
     */
    private void doApplyLog(StateMachineContext context, int index, @Nonnull byte[] commandBytes, int firstLogIndex) {
    	// 忽略已应用过的日志
        if (index <= lastApplied) {
            return;
        }
        logger.info("apply log {}", index);
        applyCommand(commandBytes);
        // 更新lastApplied
        lastApplied = index;
        onLastAppliedAdvanced();
        // 是否要生成日志快照
        if (shouldGenerateSnapshot(firstLogIndex, index)) {
        	// 生成日志快照
            context.generateSnapshot(index);
        }
    }

    @Override
    public void advanceLastApplied(int index) {
    	taskExecutor.submit(() -> {
    		if(index <= lastApplied) {
    			return;
    		}
    		lastApplied = index;
    		onLastAppliedAdvanced();
    	});
    }

    /**lastApplied有更新时执行此方法
     */
    private void onLastAppliedAdvanced() {
    	logger.info("last applied index advanced, {}", lastApplied);
    	// 取得比lastApplied小或者等于readIndex以及对应的请求
    	SortedMap<Integer, List<String>> subMap = readIndexMap.headMap(lastApplied + 1);
    	for (List<String> requestIds : subMap.values()) {
			for (String requestId : requestIds) {
				// 应用请求
				onReadIndexReached(requestId);
			}
		}
    	// 删除这些请求
    	subMap.clear();
	}

	/**应用命令(核心组件回调上层组件)
     * @param commandBytes
     */
    protected abstract void applyCommand(@Nonnull byte[] commandBytes);

    // run in node thread
    @Override
    public void applySnapshot(@Nonnull Snapshot snapshot) throws IOException {
        logger.info("apply snapshot, last included index {}", snapshot.getLastIncludedIndex());
        doApplySnapshot(snapshot.getDataStream());
        lastApplied = snapshot.getLastIncludedIndex();
    }

    protected abstract void doApplySnapshot(@Nonnull InputStream input) throws IOException;

    @Override
    public void shutdown() {
        try {
            taskExecutor.shutdown();
        } catch (InterruptedException e) {
            throw new StateMachineException(e);
        }
    }

    @Override
	public void onReadIndexReached(String requestId, int readIndex) {
    	logger.info("read index reached, request id {}, read index {}", requestId, readIndex);
    	// 在服务状态机的线程中执行
    	taskExecutor.submit(() -> {
    		if(lastApplied >= readIndex) {
    			// 如果lastApplied已经达到readIndex的位置，则直接调用读取服务的实现方法
    			onReadIndexReached(requestId);
    		}else {
    			// 否则加入等待映射
    			logger.info("waiting for last applied index {} to reach read index {}", lastApplied, readIndex);
    			// 一个ReadIndex可能对应多个请求，链表是为了节省空间以及加快添加
    			List<String> requestIds = readIndexMap.get(requestId);
    			if(requestIds == null) {
    				requestIds = new LinkedList<>();
    			}
    			requestIds.add(requestId);
    			readIndexMap.put(readIndex, requestIds);
    		}
    	});
	}

    /**kv服务中get方法的实现，这里不需要传入readIndex，也就是说，对kv服务的状态机来说readIndex是透明的。服务的状态机应该在此方法中实现读取的逻辑，对于kv服务来说就是get命令
     * @param requestId
     */
    protected abstract void onReadIndexReached(String requestId);

}
